package bean.sink

import bean.Case.CatergoryLogWide
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}

import java.sql.{Connection, DriverManager, PreparedStatement}

/**
 *
 * @program: bean.sink
 * @author: YCLW058
 * @create: 2021-06-01 11:37
 * @decsription:
 *
 * */
class Sink1() extends RichSinkFunction[CatergoryLogWide] {
  // 定义连接 预编译语句
  var conn: Connection = _
  var ps: PreparedStatement = _


  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/cd_mysql", "root", "root")
    val sql = "INSERT INTO `app`(`categoryID`,`produceID`,`userID`,`price`,`country`,`network`,`source`,`buyType`,`buyTime`,`leaveTime`,`count`,`timestamp`)VALUES (?,?,?,?,?,?,?,?,?,?,?,?)"
    ps = conn.prepareStatement(sql)

  }

  override def invoke(value: CatergoryLogWide, context: SinkFunction.Context): Unit = {
    //获取 具体值
    ps.setLong(1, value.categoryID)
    ps.setLong(2,value.produceID)
    ps.setLong(3,value.userID)
    ps.setDouble(4,value.price)
    ps.setString(5,value.country)
    ps.setString(6,value.network)
    ps.setString(7,value.source)
    ps.setString(8,value.buyType)
    ps.setLong(9,value.buyTime)
    ps.setLong(10,value.leaveTime)
    ps.setLong(11,value.count)
    ps.setLong(12,value.timestamp)

    //执行
    ps.executeLargeUpdate()

  }


  override def close(): Unit = {
    if (conn != null) conn.close()
    if (ps != null) ps.close()
  }
}
